package com.bw.gmall.app.DWD;


/*
1.获取ods 中日志数topic_log
2.处理过滤topic_log
3.对新老用户进行处理
               is_news:1
                         状态=null  我要把当前人ts转换成年月日  跟新到状态值
                         状态！=null   判断当前数据的ts 和状态日期是否相同  如果不同跟新数据
               is_new:0
                         状态=null    将ts-1day 跟新到状态中
                         如果有值 不需要处理
4. 把主流数据拆分成5个流   分别存入到5个主题中
1.创建流式环境
2.设置并行度
3.运行jar 向topic_log主题 发送数据
4.从 Kafka 读取主流数据
5.数据清洗，转换结构
6.将脏数据写出到 Kafka 指定主题
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.utils.DateFormatUtil;
import com.bw.gmall.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import static java.sql.DriverManager.println;

public class BaseLogApp {
    public static void main(String[] args) throws Exception {
//TODO 1.初始化
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //启动状态后端
//TODO 2.从kafka 中读取数据
        String topic="topic_log";
        String groupId="base_log_app";
        DataStreamSource<String> kafkaSource = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        //打印数据
//        kafkaSource.print();
//        println("打印数据");
// TODO 3.数据清洗  转换结构
        OutputTag<String> dirtyStream = new OutputTag<String>("dirtyStream") {
        };
        SingleOutputStreamOperator<String> cleanedStream = kafkaSource.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String jsonStr, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                try {
                    // 解析JSON字符串以验证其有效性，虽然当前代码没有使用解析结果，但这是数据清洗的重要步骤
                    JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                    if (jsonObject != null) {
                        JSONObject common = jsonObject.getJSONObject("common");
                        long ts = jsonObject.getLongValue("ts");
                        if (common != null && ts != 0){
                            out.collect(jsonStr);
                        }
                    }

                } catch (Exception e) {
                    // 记录脏数据到侧输出流
                    ctx.output(dirtyStream, jsonStr);
                }
            }
        });
        
// 获取并打印脏数据流
        DataStream<String> dirtyDataStream = cleanedStream.getSideOutput(dirtyStream);
//        dirtyDataStream.print();
//        cleanedStream.print();


//TODO 4.转换为json格式
        //转换主流数据结构 jsonStr -> jsonObj
        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSON::parseObject);
//        mappedStream.print();




// TODO 5. 新老访客状态标记修复   状态  keyby  mid
        //首先对mid进行分组
        KeyedStream<JSONObject, String> midStream = mappedStream.keyBy(r -> r.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> isNew = midStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    //创建状态   给相同的mid共享   容器中只存日期   value  list map
                    ValueState<String> firstViewDtState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState", String.class));
                    }

                    @Override
                    public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                        // 安全地获取 common 对象，避免空指针异常
                        JSONObject common = value.getJSONObject("common");
                        if (common == null) {
                            // 如果 common 对象不存在，直接输出原始数据
                            out.collect(value);
                            return;
                        }

                        String isNew = common.getString("is_new");
                        //状态时间
                        String firstViewDt = firstViewDtState.value();
                        //获取当前数据时间
                        Long ts = value.getLong("ts");

                        // 检查时间戳是否存在
                        if (ts == null) {
                            // 如果时间戳不存在，直接输出原始数据
                            out.collect(value);
                            return;
                        }

                        //把时间戳转化为年月日
                        String s = DateFormatUtil.toDate(ts);
                        if ("1".equals(isNew)) {//新用户
                            if (firstViewDt == null) {//没有注册时间  新用户
                                firstViewDtState.update(s);
                            } else {//有注册时间                    误标为新用户
                                if (!firstViewDt.equals(s)) {
                                    isNew = "0";
                                    common.put("is_new", isNew);
                                }
                            }
                        } else if ("0".equals(isNew)) {//老用户
                            if (firstViewDt == null) {//没有注册时间的老用户
                                String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                firstViewDtState.update(yesterday);
                            }
                        }
                        // 如果 isNew 既不是 "1" 也不是 "0"，则不做特殊处理，直接输出

                        out.collect(value);
                    }
                }
        );
//        isNew.print();

////TODO 6.分流处理    page action displays error start
//        //创建测流
        OutputTag<String> actionOutput = new OutputTag<String>("actionOutput") {};
        OutputTag<String> displaysOutput = new OutputTag<String>("displaysOutput") {};
        OutputTag<String> errorOutput = new OutputTag<String>("errorOutput") {};
        OutputTag<String> startOutput = new OutputTag<String>("startOutput") {};

        //分流
        SingleOutputStreamOperator<String> flProcess = isNew.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) throws Exception {
                        //错误日志
                        JSONObject err = value.getJSONObject("err");
                        if (err != null) {
                            ctx.output(errorOutput, value.toJSONString());
                        }
                        value.remove("err");
                        //启动日志
                        JSONObject start = value.getJSONObject("start");
                        if (start != null) {
                            ctx.output(startOutput, value.toJSONString());
                        } else {
                            JSONObject page = value.getJSONObject("page");
                            JSONObject common = value.getJSONObject("common");
                            long ts = value.getLongValue("ts");
                            value.remove("action");
                            //曝光日志
                            JSONArray displays = value.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    JSONObject json1 = new JSONObject();
                                    json1.put("display", display);
                                    json1.put("page", page);
                                    json1.put("common", common);
                                    json1.put("ts", ts);
                                    ctx.output(displaysOutput, json1.toJSONString());
                                }
                            }
                            value.remove("displays");
                            //动作日志
                            JSONArray action = value.getJSONArray("action");
                            if (action != null && page != null && common != null && ts != 0) {
                                for (int i = 0; i < action.size(); i++) {
                                    JSONObject action1 = action.getJSONObject(i);
                                    JSONObject json1 = new JSONObject();
                                    json1.put("action", action1);
                                    json1.put("page", page);
                                    json1.put("common", common);
                                    json1.put("ts", ts);
                                    ctx.output(actionOutput, json1.toJSONString());
                                }
                            }
                            value.remove("action");
                            //page 日志
                            if (page != null && common != null && ts != 0) {
                                out.collect(value.toJSONString());
                            }

                        }
                    }
                }
        );

        DataStream<String> action_Tag = flProcess.getSideOutput(actionOutput);
        DataStream<String> error_Tag = flProcess.getSideOutput(errorOutput);
        DataStream<String> displays_Tag = flProcess.getSideOutput(displaysOutput);
        DataStream<String> start_Tag = flProcess.getSideOutput(startOutput);



       // 7.2 定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        flProcess.print("page>>>>>>>>>>>>>>>>>>>");
        flProcess.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        start_Tag.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displays_Tag.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        action_Tag.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        error_Tag.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        // 执行Flink作业，添加作业名称便于识别
        env.execute("BaseLogApp");
    }

}
